package com.millstein.realtime.app.dwd.db;

import com.millstein.realtime.app.base.BaseSqlApp;
import com.millstein.realtime.common.Constants;
import com.millstein.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Description
 * @Author tsing
 * @Date 2024-10-09 13:57
 */
public class Dwd_04_TradeCartAdd extends BaseSqlApp {

    public static void main(String[] args) {
        new Dwd_04_TradeCartAdd().init(
                5000,
                3,
                "Dwd_04_TradeCartAdd"
        );
    }

    /**
     * 具体数据处理的逻辑，由子类编写
     *
     * @param env      执行环境
     * @param tableEnv 表执行环境
     */
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1.从kafka中读取ods层的数据，即topic_db中的数据
        readOdsDataFromKafka(tableEnv, "Dwd_04_TradeCartAdd");

        // 2.从mysql中读取base_dic表中的数据作为维度表，用于lookup join
        readBaseDicFromMysql(tableEnv);

        // 3.过滤出加购物车的数据
        Table cartTable = tableEnv.sqlQuery(
                "select " +
                "    `data`['id'] id, " +
                "    `data`['user_id'] user_id, " +
                "    `data`['sku_id'] sku_id, " +
                "    if ( " +
                "        `type` = 'update', " +
                "        (cast(`data`['sku_num'] as int) - cast(`old`['sku_num'] as int)), " +
                "        cast(`data`['sku_num'] as int) " +
                "    ) sku_num, " +
                "    `data`['source_type'] source_type, " +
                "    `data`['source_id'] source_id, " +
                "    `ts`, " +
                "    `pt` " +
                "from maxwell_table " +
                "where " +
                "    `table` = 'cart_info' " +
                "    and ( " +
                "        `type` = 'insert' " +
                "        or `type` = 'bootstrap-insert' " +
                "        or (`type` = 'update' " +
                "            and `old`['sku_num'] is not null " +
                "            and (cast(`data`['sku_num'] as int) - cast(`old`['sku_num'] as int)) > 0) " +
                "    ) "
        );
        tableEnv.createTemporaryView("cart_add_info", cartTable);

        // 4.维度退化
        Table resultTable = tableEnv.sqlQuery(
                "select" +
                "    cai.id, " +
                "    cai.user_id, " +
                "    cai.sku_id, " +
                "    cai.sku_num, " +
                "    cai.source_type, " +
                "    bd.dic_name source_name, " +
                "    ts " +
                "from cart_add_info cai " +
                "left join base_dic for system_time as of cai.pt as bd " +
                "    on cai.source_type = bd.dic_code "
        );

        // 5.将维度退化后的数据写到kafka中
        tableEnv.executeSql(
                "create table dwd_trade_cart_add ( " +
                "    id string, " +
                "    user_id string, " +
                "    sku_id string, " +
                "    sku_num int, " +
                "    source_type string, " +
                "    source_type_name string, " +
                "    ts bigint " +
                ")" + SqlUtil.getKafkaSinkDDL(Constants.TOPIC_DWD_TRADE_CART_ADD)
        );
        resultTable.executeInsert("dwd_trade_cart_add");
    }
}
